package com.hub.realtime.resource.service.impl;


import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hub.realtime.common.core.domain.model.ClusterInfo;
import com.hub.realtime.common.core.domain.model.LoginUser;
import com.hub.realtime.common.exception.UtilException;
import com.hub.realtime.common.utils.DateUtils;
import com.hub.realtime.common.utils.FlinkUtils;
import com.hub.realtime.common.utils.StringUtils;
import com.hub.realtime.common.utils.hadoop.HadoopUtil;
import com.hub.realtime.framework.service.ServiceBase;
import com.hub.realtime.resource.domain.ResApplication;
import com.hub.realtime.resource.domain.ResCluster;
import com.hub.realtime.resource.dto.input.FlinkVersionQueryParam;
import com.hub.realtime.resource.mapper.ResFlinkVersionsMapper;
import org.springframework.stereotype.Service;
import com.hub.realtime.resource.domain.ResFlinkVersions;
import com.hub.realtime.resource.service.IResFlinkVersionsService;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * Flink版本信息Service业务层处理
 *
 * @author wupeng
 * @date 2022-01-07
 */
@Service
public class ResFlinkVersionsServiceImpl extends ServiceBase<ResFlinkVersionsMapper, ResFlinkVersions> implements IResFlinkVersionsService {

    @Override
    public Page<ResFlinkVersions> list(FlinkVersionQueryParam param) {
        LambdaQueryWrapper<ResFlinkVersions> wrapper = new LambdaQueryWrapper<>();
        wrapper.orderByDesc(ResFlinkVersions::getCreateTime);
        if (StringUtils.isNotEmpty(param.getFlinkVersion())) {
            wrapper.like(ResFlinkVersions::getFlinkVersion, param.getFlinkVersion());
        }
        return getPageList(param, wrapper);
    }

    @Override
    public List<ResFlinkVersions> allList() {
        LambdaQueryWrapper<ResFlinkVersions> wrapper = new LambdaQueryWrapper<>();
        wrapper.orderByDesc(ResFlinkVersions::getCreateTime);
        return baseMapper.selectList(wrapper);
    }

    @Override
    public int saveInfo(ResFlinkVersions param) {
        List<String> surportVersions = new ArrayList<>();
        surportVersions.add("1.12.2");
        surportVersions.add("1.13.2");
        surportVersions.add("1.14.2");
        int res = -1;
        LoginUser loginUser = getLoginUser();
        String flinkPath = param.getLocalPath();
        String flinkVersion = FlinkUtils.flinkVersion(flinkPath);
        if (!surportVersions.contains(flinkVersion)) {
            throw new UtilException("暂时不支持改版本：" + flinkVersion + " ,现在支持的版本：{" + StringUtils.join(surportVersions.toArray(), ",") + "}");
        }
        LambdaQueryWrapper<ResFlinkVersions> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ResFlinkVersions::getLocalPath, flinkPath);
        ResFlinkVersions compareParam = baseMapper.selectOne(queryWrapper);
        //查询当前路径是否存在
        if (compareParam != null) {
            //修改
            res = doSave(param, compareParam, flinkVersion, flinkPath, loginUser.getUsername(), 2);
        } else {
            //查看版本号
            queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(ResFlinkVersions::getFlinkVersion, flinkVersion);
            compareParam = baseMapper.selectOne(queryWrapper);
            if (compareParam != null) {
                res = doSave(param, compareParam, flinkVersion, flinkPath, loginUser.getUsername(), 2);
            } else {
                res = doSave(param, compareParam, flinkVersion, flinkPath, loginUser.getUsername(), 1);
            }
        }
        return res;
    }

    @Override
    public int deleteInfos(Long[] ids) {
        List<ResFlinkVersions> resFlinkVersions = baseMapper.selectBatchIds(Arrays.stream(ids).collect(Collectors.toList()));
        boolean hasInuse = false;
        if (resFlinkVersions != null && !resFlinkVersions.isEmpty()) {
            for (ResFlinkVersions flinkVersions : resFlinkVersions) {
                if (flinkVersions.isInUse()) {
                    hasInuse = true;
                    break;
                }
            }
        }
        if (hasInuse) {
            throw new UtilException("存在正在使用的版本，不允许删除，请选择未使用的版本进行删除！");
        }
        //清除集群缓存数据
        return delByIds(ids);
    }


    private int doSave(ResFlinkVersions param, ResFlinkVersions compareParam, String flinkVersion, String flinkPath, String userName, int type) {
        int res = -1;
        if (type == 1) {
            if (param.getId() != null) {
                param.setFlinkVersion(flinkVersion);
                param.setLocalPath(flinkPath);
                param.setUpdateBy(userName);
                param.setUpdateTime(DateUtils.getNowDate());
                res = baseMapper.updateById(param);
            } else {
                param.setFlinkVersion(flinkVersion);
                param.setLocalPath(flinkPath);
                param.setCreateBy(userName);
                param.setCreateTime(DateUtils.getNowDate());
                res = baseMapper.insert(param);
            }
        } else {
            if (param.getId() != null) {
                if (!Objects.equals(compareParam.getId(), param.getId())) {
                    throw new UtilException("数据库中已经存在当前版本：" + flinkVersion);
                } else {
                    param.setFlinkVersion(flinkVersion);
                    param.setLocalPath(flinkPath);
                    param.setUpdateBy(userName);
                    param.setUpdateTime(DateUtils.getNowDate());
                    res = baseMapper.updateById(param);
                }
            } else {
                throw new UtilException("数据库中已经存在当前版本：" + flinkVersion);
            }

        }


        return res;
    }

}

